package function;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;
import java.util.*;

public class RichMapSqlFunction extends RichMapFunction<Tuple2<Boolean, Row>,String> {
    List<String> l = new ArrayList<String>();
    Map<String,String> m = new LinkedHashMap<String,String>();
    String sql = "";
    @Override
    public void close() throws Exception {
        Iterator<String> iterator = m.keySet().iterator();
        while(iterator.hasNext()){
            l.add(m.get(iterator.next()));
        }
        for(int i=0;i<l.size();i++){

            if(i<1){
                sql+=l.get(i)+"\n";
            }else{
                sql+=","+l.get(i)+"\n";
            }
        }
        sql+=",w_update_dt string COMMENT '业务更新时间' \n" +
                ",w_insert_dt string COMMENT '业务插入时间' \n" +
                ",batch_dt string COMMENT '数据批次时间'";

        System.out.println(sql);
    }

    @Override
    public String map(Tuple2<Boolean, Row> value) throws Exception {
        Row f1 = value.f1;
        String field_1 = f1.getField(0).toString();
        String field_2 = f1.getField(1).toString();
        String field_3="''";
        if(f1.getField(2)!=null){
            field_3 = f1.getField(2).toString();
        }
        String str = field_1+" "+field_2+" "+"comment"+" "+field_3;
        //l.add(str);
        m.put(field_1,str);
        return "";
    }
}
